package com.link.serve.kafka.coder;
import com.google.protobuf.InvalidProtocolBufferException;
import com.link.serve.kafka.proto.KafkaMessageProto;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;

@Slf4j
public class KafkaMessageDeserializer implements Deserializer<KafkaMessageProto.KafkaMessage> {

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {}

    @Override
    public KafkaMessageProto.KafkaMessage deserialize(String topic, byte[] data) {
        try {
            return KafkaMessageProto.KafkaMessage.parseFrom(data);
        } catch (InvalidProtocolBufferException e) {
            log.error("❌Deserialization_error_topic{}:{}", topic, e.getMessage());
            throw new RuntimeException("Failed to deserialize KafkaMessage", e);
        }
    }

    @Override
    public void close() {}
}

